package com.eugenes.concurrency.sort;

import com.eugenes.concurrency.runnables.Merge;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class MergeSort< T extends Comparable<T> > {

    private static final int LIMIT = 1000;

    private List<T> data;
    private List<T> cache;

    // Current implementation make all actions with isRunning in
    // synchronized blocks. Make it volatile when necessary.
    private /* volatile */ boolean isRunning;

    // No need to do isFinished volatile because it's needed only in case when
    // sort is done without calling executor.submit() so sort is done in one thread.
    // Otherwise another thread should wait for finishedNotifier's monitor which is free
    // only in case when main thread is waiting for notify.
    private boolean isFinished;

    private List<Task> tasks;

    private final Object finishedNotifier = new Object();
    private final ThreadPoolExecutor executor;

    public MergeSort(int threads, int capacity) {
        this.tasks = new ArrayList<>();
        executor = new ThreadPoolExecutor(
                threads, threads, 1, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(capacity));
    }

    private <E> List<E> getNewNullList(int size) {
        List<E> result = new ArrayList<>(size);
        for (int i=0; i<size; ++i) {
            result.add(null);
        }
        return result;
    }

    private synchronized void checkRunning(boolean expected) {
        if (isRunning != expected) {
            throw new IllegalStateException("Illegal state: expected " +
                    (expected ? "running" : "not running"));
        }
    }

    private synchronized void startRunning() {
        checkRunning(false);
        isRunning = true;
    }

    private synchronized void stopRunning() {
        checkRunning(true);
        isRunning = false;
    }

    public void finish() {
        executor.shutdown();
    }

    private synchronized void checkData() {
        if (data == null) {
            throw new IllegalStateException("Call set data before sorting array");
        }
    }

    public synchronized void setData(Collection<? extends T> data) {
        checkRunning(false);
        this.data = new ArrayList<>(data.size());
        for (T item : data) {
            this.data.add(item);
        }
    }

    private void recursiveTasksCreation(int v, int l, int r) {
        Task mergeTask = new Task(v, l, r);
        tasks.set(v, mergeTask);
        if (r - l < LIMIT) {
            Collections.sort(data.subList(l, r + 1));
            mergeTask.notifyParent();
        } else {
            int med = (l + r) >> 1;
            recursiveTasksCreation(v+v, l, med);
            recursiveTasksCreation(v+v+1, med+1, r);
        }
    }

    private void createTasks() {
        tasks = getNewNullList(data.size() * 4);
        recursiveTasksCreation(1, 0, data.size() - 1);
    }

    public synchronized List<T> sort() throws InterruptedException {
        checkData();
        startRunning();
        cache = getNewNullList(data.size());

        synchronized (finishedNotifier) {
            isFinished = false;
            createTasks();
            while (!isFinished) {
                finishedNotifier.wait();
            }
        }

        final List<T> result = data;
        data = null;
        stopRunning();
        return result;
    }

    private class Task implements Runnable {

        private final int v;
        private final int left;
        private final int right;
        private int ready;

        public Task(int v, int left, int right) {
            this.v = v;
            this.left = left;
            this.right = right;
        }

        public synchronized void incReady() {
            ++ready;
        }

        public synchronized boolean isReady() {
            // Two parts of this task is already sorted
            return ready == 2;
        }

        private void notifyParent() {
            if (v==1) {
                isFinished = true;
                synchronized (finishedNotifier) {
                    finishedNotifier.notifyAll();
                }
            } else {
                Task parentTask = tasks.get(v/2);
                parentTask.incReady();
                if (parentTask.isReady()) {
                    executor.submit(parentTask);
                }
            }
        }

        @Override
        public void run() {
            int med = (left + right) >> 1;
            new Merge<>(data, left, med - left + 1, right - med, cache, left).run();
            notifyParent();
        }
    }
}
